package com.test;

import java.io.IOException;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.ListIterator;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.time.DateFormatUtils;
import org.apache.commons.lang.time.DateUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.nodes.Element;
import org.jsoup.select.Elements;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import us.codecraft.xsoup.Xsoup;
import us.codecraft.xsoup.xevaluator.XElements;

import com.chebao.exception.GpsException;
import com.chebao.util.GpsUtils;
import com.chebao.util.Point;


public class GpsParseHdfs2Hbase {
	private static final Logger LOG = LoggerFactory.getLogger(GpsParseHdfs2Hbase.class);
	
	//截取boxId正则
	private static final Pattern PATTERN = Pattern.compile(".*?(\\d{15}).*");
	//解析GPS标志位的XPATH
	private static final String XPATH = "//div/@style";
	private static final String DATE_PATTERN = "yyyy-MM-dd HH:mm:ss";
	private static final String DATE_FORMAT_PATTERN = "yyyyMMddHHmmss";
	
	private static final String OUTPUT_TABLE_KEY = "OUTPUT_TABLE";

	static class GpsParseMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable, Put>
	{
//		private Connection conn;
//		private Table hTable;
		@Override
		protected void setup(
				Mapper<LongWritable, Text, ImmutableBytesWritable, Put>.Context context)
				throws IOException, InterruptedException {
			super.setup(context);
			// TODO Auto-generated method stub
//			Configuration conf = context.getConfiguration();
//			String outputTableName = conf.get(OUTPUT_TABLE_KEY);
//			conn = ConnectionFactory.createConnection(conf);
//			hTable = conn.getTable(TableName.valueOf(outputTableName));
		}

		@Override
		protected void map(
				LongWritable offset,
				Text line,
				Mapper<LongWritable, Text, ImmutableBytesWritable, Put>.Context context)
				throws IOException, InterruptedException {
			
			String htmlStr = line.toString();
			
//			List<Put> puts = new  ArrayList<Put>();
			Document doc = Jsoup.parse(htmlStr);
			XElements selList = Xsoup.select(doc, XPATH);
			Elements elements = selList.getElements();
			ListIterator<Element> divIter = elements.listIterator();
			while (divIter.hasNext())
			{
				Element dataEle = divIter.next();
				String typeText = dataEle.text();
				
				if (StringUtils.contains(typeText, "接收类型-68"))
				{
					LOG.debug("=====msgType={}", typeText);
					
					Element boxIdEle = dataEle.nextElementSibling();
					String boxIdEleText = boxIdEle.text();
					String boxIdStr = grepBoxId(boxIdEleText);
					
					Element locDivEle = dataEle.nextElementSibling().nextElementSibling().nextElementSibling();
					XElements locUlXEle = Xsoup.select(locDivEle, "//ul//li");
					Elements locLiEle = locUlXEle.getElements();
					
					Element timeEle = locLiEle.get(1);
					//格式范例======时间:2015-11-20 07:31:48
					String timeText = timeEle.text();
					String timeStr = timeText.substring(3);
					LOG.debug("=====timeStr={}", timeStr);
					
					int liSize = locLiEle.size();
					if (liSize > 2)
					{
						Date date = null;
						try {
							date = DateUtils.parseDate(timeStr, new String[]{DATE_PATTERN});
						} catch (ParseException e) {
							continue;
						}
						for (int i = 2; i < liSize; i++)
						{
							String coordinate = locLiEle.get(i).text();
							Point point = null;
							try {
								point = GpsUtils.parseGps(coordinate);
								if (null == point)
									continue;
							} catch (GpsException e) {
								continue;
							}
							String dateStr = DateFormatUtils.format(date, DATE_FORMAT_PATTERN);
							String rowkey = StringUtils.join(new String[]{boxIdStr, dateStr}, "_");
							LOG.debug("========={}========{}", rowkey, point.toString());
							byte[] rowkeyBArr = Bytes.toBytes(rowkey);
							byte[] familyBArr = Bytes.toBytes("d");
							byte[] lngColArr = Bytes.toBytes("lng");
							byte[] lngBArr = Bytes.toBytes(point.getLng());
							byte[] latColArr = Bytes.toBytes("lat");
							byte[] latBArr = Bytes.toBytes(point.getLat());
							Put put = new Put(rowkeyBArr);
							put.addColumn(familyBArr, lngColArr, lngBArr);
							put.addColumn(familyBArr, latColArr, latBArr);
//							puts.add(put);
							
							//每个GPS
							date = DateUtils.addSeconds(date, 5);
							
							context.write(new ImmutableBytesWritable(Bytes.toBytes(offset + "")), put);
						}
					}
					
					
				}
			}
		
//			hTable.put(puts);
//			context.write(null, null);
			
//			context.write(new ImmutableBytesWritable(Bytes.toBytes(offset + "")), put);  
//			context.getCounter(counterName)
//			// TODO Auto-generated method stub
//			super.map(key, value, context);
		}

		@Override
		protected void cleanup(
				Mapper<LongWritable, Text, ImmutableBytesWritable, Put>.Context context)
				throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			super.cleanup(context);
//			IOUtils.closeStream(hTable);
//			IOUtils.closeStream(conn);
		}
		
	}
	
	public static String grepBoxId(String boxIdText)
	{
		if (StringUtils.isBlank(boxIdText))
		{
			return "";
		}
		Matcher matcher = PATTERN.matcher(boxIdText);
		if (matcher.matches())
		{
			String boxId = matcher.group(1);
			return  boxId;
		}
		return "";
	}
	
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException
	{
		Configuration conf = new  Configuration();
	    String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
	    int argsLen = otherArgs.length;
	    System.out.println("argsLen===" + argsLen);
	    for (int i = 0; i < argsLen; i++)
	    {
	    	System.out.println("otherArgs[" + i + "]=" + otherArgs[i]);
	    }
	    if (otherArgs.length != 5) {
	      System.err.println("Usage: GpsParseHdfs2Hbase <in> <outHbase>");
	      System.exit(2);
	    }
	    String inputPathStr = otherArgs[3];
	    String tableNameStr = otherArgs[4];
//	    conf.set(OUTPUT_TABLE_KEY, tableNameStr);
	    
		
		Job job= Job.getInstance(conf, "GpsParseHdfs2Hbase");
		job.setJarByClass(GpsParseHdfs2Hbase.class);
		job.setMapperClass(GpsParseMapper.class);
		job.setOutputFormatClass(TableOutputFormat.class);
		job.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, tableNameStr);
		job.setOutputKeyClass(ImmutableBytesWritable.class);
		job.setOutputValueClass(Put.class);
		job.setMapOutputKeyClass(ImmutableBytesWritable.class);
		job.setMapOutputValueClass(Put.class);
		job.setNumReduceTasks(0);
		FileInputFormat.addInputPath(job, new Path(inputPathStr));
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
}
